#include <errno.h>
#include <netdb.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include "compat.h"

#include "psd_modules.h"
#include "psd_session.h"
#include "psd_config.h"
#include "psd_event.h"
#include "psd_measurement.h"
#include "psd_logger.h"
#include "psd_tpool.h"

psdModule *module_info();
int psd_stats_init();
static void *psd_stats_listener_handler(void *v);
static void psd_stats_read_config();
static int psd_write_stats(int sd);
static int psd_print_stats(char *buf, int bufsize);
static int psd_session_print_stats(char *buf, int bufsize, psdSess *session);
static int psd_session_print_final_stats(char *buf, int bufsize, psdSess *session);
static int psd_conn_print_stats(char *buf, int bufsize, psdConn *conn);
static int psd_write_history_stats(int sd);
static int psd_stats_session_end_cb(psdSess *sess);

static int *sockets;
static int num_sockets;

static char **session_history;
static int session_history_count;

static psdModule psd_stats_module = {
	.desc = "PSD Stats Daemon Module",
	.dependencies = "",
	.init = psd_stats_init
};

struct psd_stats_config_t {
	unsigned int port;
};

static struct psd_stats_config_t psdStatsConfig = {
	.port = 5009
};

psdModule *module_info() {
	return &psd_stats_module;
}

int psd_stats_init() {
	int i;
	struct addrinfo *ai;

	psd_stats_read_config();

	sockets = listen_port(IPPROTO_TCP, PF_UNSPEC, psdStatsConfig.port, &num_sockets, &ai);
	if (!sockets) {
		psd_err(0, "couldn't listen on port: %d", psdStatsConfig.port);
		goto error_exit;
	}

	freeaddrinfo(ai);

	session_history = NULL;
	session_history_count = 0;

	if (register_listener_session_end(psd_stats_session_end_cb)) {
		psd_err(0, "couldn't register stats callback");
		goto error_exit2;
	}

	if (psd_tpool_exec(psd_stats_listener_handler, NULL) != 0) {
		psd_err(0, "couldn't execute socket handler thread");
		goto error_exit2;
	}

	return 0;

error_exit2:
	for(i = 0; i < num_sockets; i++)
		close(sockets[i]);
error_exit:
	return -1;
}

static int psd_stats_session_end_cb(psdSess *sess) {
	char buf[2048];

	if (psd_session_print_final_stats(buf, sizeof(buf), sess) == 0) {
		strlist_add(buf, &session_history, &session_history_count);
	}

	return 0;
}

static int psd_session_print_final_stats(char *buf, int bufsize, psdSess *session) {
	uint64_t bytes_read;
	size_t bytes_read_size;
	uint64_t bytes_written;
	size_t bytes_written_size;
	double timelen;
	psdConn *curr_conn;
	char tmp[256];
	int n;

	buf[0] = '\0';

	strlcat(buf, psd_session_get_id(session), bufsize);

	strlcat(buf, ",", bufsize);
	for(curr_conn = session->parent_conns; curr_conn != NULL; curr_conn = curr_conn->next) {
		strlcat(buf, curr_conn->description, bufsize);
		if (curr_conn->next != NULL)
			strlcat(buf, "|", bufsize);
	}

	strlcat(buf, ",", bufsize);
	for(curr_conn = session->child_conns; curr_conn != NULL; curr_conn = curr_conn->next) {
		strlcat(buf, curr_conn->description, bufsize);
		if (curr_conn->next != NULL)
			strlcat(buf, "|", bufsize);
	}

	if (psd_session_get_stat(session, PSD_STATS_BYTES_READ, &bytes_read, &bytes_read_size) != 0) {
		bytes_read = 0;
	}

	if (psd_session_get_stat(session, PSD_STATS_BYTES_WRITTEN, &bytes_written, &bytes_written_size) == 0) {
		bytes_written = 0;
	}

	timelen = difftv(&(session->start_time), &(session->end_time));
	snprintf(tmp, sizeof(tmp), ",%f,%llu,%llu,%f,%f", timelen, bytes_read, bytes_written, bytes_read/timelen*8/1000/1000, bytes_written/timelen*8/1000/1000);
	n = strlcat(buf, tmp, bufsize);

	if (n > bufsize)
		return -1;
	else
		return 0;
}

static void psd_stats_read_config() {
	int val;

	if (psd_depot_settings_get_int("stats", "port", &val) == 0) {
		if (val >= 0)
			psdStatsConfig.port = val;
	}
}

static void *psd_stats_listener_handler(void *v) {
	fd_set socket_set;
	int n, i, high;
	int sd;

	pthread_detach(pthread_self());

	while(1) {

		FD_ZERO(&socket_set);

		high = 0;

		for(i = 0; i < num_sockets; i++) {
			FD_SET(sockets[i], &socket_set);
			if (sockets[i] > high)
				high = sockets[i];
		}

		n = select(high + 1, &socket_set, NULL, NULL, NULL);

		for(i = 0; i < num_sockets; i++) {
			if (FD_ISSET(sockets[i], &socket_set)) {
				struct sockaddr_storage sa;
				SOCKLEN_T sa_len = sizeof(struct sockaddr_storage);

				sd = accept(sockets[i], (struct sockaddr *) &sa, &sa_len);
				if (sd < 0) {
					psd_err(1, "accept failed: %s", strerror(errno));
					goto error_exit;
				}

				send(sd, "history\n", 8, 0);
				psd_write_history_stats(sd);
				send(sd, "done history\n", 13, 0);
				send(sd, "curr\n", 5, 0);
				psd_write_stats(sd);
				send(sd, "done curr\n", 10, 0);

				close(sd);
			}
		}
	}

error_exit:
	return NULL;

}

static int psd_write_history_stats(int sd) {
	int i, n;

	for(i = 0; i < session_history_count; i++) {
		n = send(sd, session_history[i], strlen(session_history[i]), 0);
		if (n <= 0) {
			return -1;
		}
		n = send(sd, "\n", 1, 0);
		if (n <= 0) {
			return -1;
		}
	}

	return 0;
}

static int psd_write_stats(int sd) {
	char buf[2048];
	int n;

	n = psd_print_stats(buf, sizeof(buf));
	if (n > sizeof(buf)) {
		return -1;
	}

	n = send(sd, buf, n, 0);

	return n;
}

static int psd_print_stats(char *buf, int bufsize) {
	psdSess **sessions;
	int session_count;
	int i, n;
	int total = 0;

	sessions = psd_get_sessions(&session_count);
	if (!sessions) {
		goto out;
	}

	for(i = 0; i < session_count; i++) {
		n = psd_session_print_stats(buf, bufsize, sessions[i]);
		if (n < bufsize) {
			buf += n;
			bufsize -= n;
		}
		total += n;
	}

	for(i = 0; i < session_count; i++) {
		psd_session_put_ref(sessions[i]);
	}
	free(sessions);
out:
	return total;
}

static int psd_session_print_stats(char *buf, int bufsize, psdSess *session) {
	psdConn *conn;
	int n;
	int total;

	total = 0;

	for(conn = session->parent_conns; conn != NULL; conn = conn->next) {
		n = psd_conn_print_stats(buf, bufsize, conn);
		if (n < bufsize) {
			strlcat(buf, "\n", bufsize);
			buf += n + 1;
			bufsize -= n + 1;
		}
		total += n + 1;
	}

	for(conn = session->child_conns; conn != NULL; conn = conn->next) {
		n = psd_conn_print_stats(buf, bufsize, conn);
		if (n < bufsize) {
			strlcat(buf, "\n", bufsize);
			buf += n + 1;
			bufsize -= n + 1;
		}
		total += n + 1;
	}

	return total;
}

static int psd_conn_print_stats(char *buf, int bufsize, psdConn *conn) {
	float RTT;
	size_t RTT_size = sizeof(RTT);
	uint32_t bytes_retr;
	size_t bytes_retr_size = sizeof(bytes_retr);
	uint64_t bytes_sent;
	size_t bytes_sent_size = sizeof(bytes_sent);
	uint64_t bytes_read;
	size_t bytes_read_size = sizeof(bytes_read);
	int n;
	struct timeval tv;

	gettimeofday(&tv, NULL);

	if (conn->get_stat(conn, PSD_STATS_RTT, &RTT, &RTT_size) != 0) {
		RTT = 0;
	}

	if (conn->get_stat(conn, PSD_STATS_BYTES_READ, &bytes_read, &bytes_read_size) != 0) {
		bytes_read = 0;
	}

	if (conn->get_stat(conn, PSD_STATS_BYTES_WRITTEN, &bytes_sent, &bytes_sent_size) != 0) {
		bytes_sent = 0;
	}

	if (conn->get_stat(conn, PSD_STATS_BYTES_RETRANSMITTED, &bytes_retr, &bytes_retr_size) != 0) {
		bytes_retr = 0;
	}

	n = snprintf(buf, bufsize, "%u.%u,%s,%s,%s,%f,%llu,%llu,%u",
			(unsigned int) tv.tv_sec, (unsigned int) tv.tv_usec,
			psd_session_get_id(conn->session),
			conn->description, psd_session_get_user(conn->session),
			RTT, bytes_read, bytes_sent, bytes_retr);

	return n;
}
